package com.warren.financial.lease.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.warren.financial.lease.realtime.app.func.AsyncDimFunctionHBase;
import com.warren.financial.lease.realtime.bean.DwsAuditAuditManCancelBean;
import com.warren.financial.lease.realtime.bean.DwsAuditAuditManRejectBean;
import com.warren.financial.lease.realtime.util.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @AUTHOR warren
 * @date 2024/1/11
 * @Introduction
 **/
public class DwsAuditAuditManRejectWin {

    public static void main(String[] args) throws Exception {
        // 1 创建流环境
        String appName = "dws_audit_audit_man_reject_window";
        StreamExecutionEnvironment env = CreateEnvUtil.getExecutionEnvironment(8082, appName);
        env.setParallelism(1);

        // 2. 从kafka 读取数据
        String topicName = "financial_dwd_audit_reject";
        KafkaSource<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(topicName, appName, OffsetsInitializer.earliest());
        DataStreamSource<String> kafkaSource = env.fromSource(kafkaConsumer, WatermarkStrategy.noWatermarks(), "kafka_source");

        // 3. 转换结构
        SingleOutputStreamOperator<DwsAuditAuditManRejectBean> flatMapStream = kafkaSource.flatMap(new FlatMapFunction<String, DwsAuditAuditManRejectBean>() {
            @Override
            public void flatMap(String value, Collector<DwsAuditAuditManRejectBean> out) throws Exception {
                DwsAuditAuditManRejectBean bean = JSON.parseObject(value, DwsAuditAuditManRejectBean.class);
                if (bean.getAuditManId() != null) {
                    bean.setApplyCount(1L);
                    out.collect(bean);
                }
            }
        });

        //4 引入水位线
        SingleOutputStreamOperator<DwsAuditAuditManRejectBean> withWaterMark = flatMapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsAuditAuditManRejectBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner<DwsAuditAuditManRejectBean>() {
            @Override
            public long extractTimestamp(DwsAuditAuditManRejectBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        //5 业务经审办 分组
        KeyedStream<DwsAuditAuditManRejectBean, String> keyByStream = withWaterMark.keyBy(new KeySelector<DwsAuditAuditManRejectBean, String>() {
            @Override
            public String getKey(DwsAuditAuditManRejectBean value) throws Exception {
                return value.getAuditManId();
            }
        });

        //6. 开窗
        WindowedStream<DwsAuditAuditManRejectBean, String, TimeWindow> windowStream = keyByStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        //7. 聚合
        SingleOutputStreamOperator<DwsAuditAuditManRejectBean> reduceStream = windowStream.reduce(new ReduceFunction<DwsAuditAuditManRejectBean>() {
            @Override
            public DwsAuditAuditManRejectBean reduce(DwsAuditAuditManRejectBean value1, DwsAuditAuditManRejectBean value2) throws Exception {
                value1.setApplyCount(value1.getApplyCount() + value2.getApplyCount());
                value1.setApplyAmount(value1.getApplyAmount().add(value1.getApplyAmount()));
                return value1;
            }
        }, new ProcessWindowFunction<DwsAuditAuditManRejectBean, DwsAuditAuditManRejectBean, String, TimeWindow>() {
            @Override
            public void process(String s, ProcessWindowFunction<DwsAuditAuditManRejectBean, DwsAuditAuditManRejectBean, String, TimeWindow>.Context context, Iterable<DwsAuditAuditManRejectBean> elements, Collector<DwsAuditAuditManRejectBean> out) throws Exception {
                String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                String curDate = DateFormatUtil.toDate(context.window().getStart());
                for (DwsAuditAuditManRejectBean element : elements) {
                    element.setStt(stt);
                    element.setEdt(edt);
                    element.setCurDate(curDate);
                    out.collect(element);
                }
            }
        });

        //8 关联维度信息
        SingleOutputStreamOperator<DwsAuditAuditManRejectBean> auditStream = AsyncDataStream.unorderedWait(reduceStream, new AsyncDimFunctionHBase<DwsAuditAuditManRejectBean>() {
            @Override
            public String getTable() {
                return "dim_employee";
            }

            @Override
            public String getId(DwsAuditAuditManRejectBean bean) {
                return bean.getAuditManId();
            }

            @Override
            public void addDim(DwsAuditAuditManRejectBean bean, JSONObject dim) {
                bean.setAuditManName(dim.getString("name"));
            }
        }, 60, TimeUnit.SECONDS);

        // 9写出到doris
        auditStream.map(new MapFunction<DwsAuditAuditManRejectBean, String>() {
            @Override
            public String map(DwsAuditAuditManRejectBean value) throws Exception {
                return Bean2JSONUtil.bean2JSON(value);
            }
        }).sinkTo(DorisUtil.getDorisSink("financial_lease_realtime.dws_audit_audit_man_reject_win","dws_audit_audit_man_reject_win"));

        //10 执行
        env.execute();
    }
}
